package hk.linkcredit.common.rocketmq.listener;

import com.alibaba.fastjson.JSONObject;
import hk.linkcredit.common.rocketmq.util.RocketMqUtils;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@AllArgsConstructor
@Slf4j
@RocketMQTransactionListener
@ConditionalOnProperty(prefix = "rocketmq", name = "producer.group")
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private final Map<String, ITransactionHandler> transactionHandlerMap;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            log.info("处理本地事务-> message:{},object:{}", JSONObject.toJSONString(message), JSONObject.toJSONString(o));
            String msgId = message.getHeaders().get(RocketMqUtils.MSG_ID).toString();
            ITransactionHandler transactionHandler = this.getTransactionHandler(message);
            if (transactionHandler == null) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            transactionHandler.processTransaction(o, msgId);
        } catch (Exception e) {
            log.error("处理本地事务异常", e);
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        return RocketMQLocalTransactionState.COMMIT;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        try {
            log.info("检查本地事务-> message:{}", JSONObject.toJSONString(message));
            String msgId = message.getHeaders().get(RocketMqUtils.MSG_ID).toString();
            String bizId = message.getHeaders().get(RocketMqUtils.BIZ_ID).toString();
            ITransactionHandler transactionHandler = this.getTransactionHandler(message);
            if (transactionHandler == null || !transactionHandler.checkTransaction(bizId, msgId)) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            log.error("检查本地事务异常", e);
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 获取事务监听实现
     *
     * @param message
     * @return
     */
    private ITransactionHandler getTransactionHandler(Message message) {
        String beanName = message.getHeaders().get(RocketMqUtils.BEAN_NAME).toString();
        ITransactionHandler transactionHandler = transactionHandlerMap.get(beanName);
        if (transactionHandler == null) {
            log.warn("beanName[{}] is not implements [ITransactionHandler] interface", beanName);
        }
        return transactionHandler;
    }
}
